package org.sunyaxing.transflow.transflowapp.controllers;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Service;
import org.sunyaxing.transflow.transflowapp.controllers.wshandler.StringProcessingHandler;

/**
 * Agent 上报上来的消息发送到这里
 * 将数据打印到日志文件
 */
@Service
public class StreamMessageWs implements ApplicationRunner {

    private static final Logger log = LoggerFactory.getLogger(StreamMessageWs.class);
    private final EventLoopGroup boss;
    private final EventLoopGroup worker;

    @Value("${transflow.agent.server:127.0.0.1}")
    private String agentHost;
    @Value("${transflow.agent.port:18988}")
    private Integer agentPort;

    public StreamMessageWs() {
        boss = Epoll.isAvailable() ? new EpollEventLoopGroup(1) : new NioEventLoopGroup(1);
        worker = Epoll.isAvailable() ? new EpollEventLoopGroup() : new NioEventLoopGroup();
    }

    public void run() {
        try {
            log.info("【AGENT SERVER】 START AT {}:{}", agentHost, agentPort);
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(boss, worker)
                    .channel(Epoll.isAvailable() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, Boolean.TRUE)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new LineBasedFrameDecoder(1024));
                            pipeline.addLast(new StringDecoder());
                            pipeline.addLast(new StringEncoder());
                            pipeline.addLast(new StringProcessingHandler());
                        }
                    });
            ChannelFuture future = bootstrap.bind(agentHost, agentPort).sync();
            //future.channel().closeFuture().sync();
            future.channel().closeFuture().addListener((ChannelFutureListener) channelFuture -> {
                channelFuture.channel().close();
            });
            log.info("【AGENT SERVER】 START COMPLETE");
        } catch (Exception e) {
            log.error("e: ", e);
            destroy();
        } finally {
        }
    }

    public void destroy() {
        boss.shutdownGracefully();
        worker.shutdownGracefully();
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        run();
    }
}
